package cn.edu.laud.sany.dataimport;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import redis.clients.jedis.Jedis;

import cn.edu.laud.sany.types.DataFormatErrorException;
import cn.edu.laud.sany.types.ImportType;
import cn.edu.laud.sany.types.RowData;
import cn.edu.laud.sany.types.TimeSerial;
import cn.edu.laud.sany.types.UnConfiguredCFException;
import cn.edu.laud.sany.utils.Config;
import cn.edu.laud.sany.utils.MutationMap;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;
import cn.edu.laud.sany.utils.TimeUtils;

public class LaUDImport
{
	// config params
	public static int thread = 1;
	public static int schema = 1;
	public static int laudPort = Config.InitLaudPort;
	public static String[] IP = new String[] {};
	public static String[] host = new String[] {};
	public static String keyspace = "";
	public static ImportType importType;
	public static String csvFolder = "";
	public static String cfsFile = "";
	public static String ignoreParaIdsFile = "";
	public static String writeTranscation = "ANY";
	public static int replicationFactor = 2;
	public static boolean errorDataStop = false;
	public static boolean ignoreSomeParaId = false;
	public static boolean doNotWait = false;
	public static String redisIp = "192.168.10.35";
	
	// used constants
	public static Integer[] cFS;
	public static HashSet<Integer> ignoreParaIds = new HashSet<Integer>();
	public static String cfNameFormat = "";
	public static String cfPrimaryKeyAlias = "";
	public static String cfComparatorType = Config.CFComparatorType0;
	public static String logFileFolder = "log";
	public static String logFileName = "mylog.txt";
	public static String cQLInsertFormat = "";
	
	// running variables
	private static ThreadPoolExecutor executor;
	private static List<LaUDRunner> runners;
	private static MyLog logger;
	public static Boolean halted = false;
	
	// constants
	public static int PROCESS_LOG_BATCH_SIZE = 10;
	public static int READ_BATCH_SIZE = 1000;
	public static int MUTATION_BATCH_SIZE = 10000;
	public static int TIMESERIAL_MAP_INIT_CAPACITY = 10000000;
	
	// for parse data and add timeSerial
	protected static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	protected static SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
	// equipcode+paraId to TimeSerial map
	protected static Map<String, TimeSerial> timeSerialMap;
	
	// for import process
	protected static int threadNum;
	protected static String nowFileName = "";
	protected static long nowLineNum = 0;
	// protected static long nowImportCount = 0;
	public static Boolean nomoreCsv = false;
	protected static ConcurrentLinkedQueue<MutationMap> mutationMapQueue;
	public static Jedis jedis;
	
	/**
	 * @param args
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static void main(String[] args) throws ClassNotFoundException, SQLException, IOException
	{
		if (initMyLog() == false) return;
		doPrepare(args);
		while (true)
		{
			if (jedis.get("finish").trim().equals("false"))
			{
				logger.info("Start Import...");
				doImport();
			}
			else
			{
				WaitSecond(1);
			}
		}
	}
	
	public static String getLogFilePath()
	{
		return (logFileFolder + "/" + logFileName);
	}
	
	protected static boolean initMyLog()
	{
		File logFolderFile = new File(logFileFolder);
		if (logFolderFile.exists() == false)
		{
			try
			{
				logFolderFile.mkdir();
			}
			catch (Exception e)
			{
				System.err
						.println("Can not create log folder, please check your permission settings!");
				return false;
			}
		}
		logger = MyLog.getMyLog(getLogFilePath(), true);
		return true;
	}
	
	protected static void doPrepare(String[] args) throws ClassNotFoundException, SQLException,
			IOException
	{
		if (args.length % 2 == 1 || args.length < 1)
		{
			System.err.println(Config.CommandHelpString);
			return;
		}
		
		try
		{
			parseParams(args);
		}
		catch (Exception e)
		{
			System.err.println(Config.CommandHelpString);
			return;
		}
		
		// test csv and cfs files
		if (testCsvFoler() == false)
		{
			System.err.println("Can not find csv folder!");
			return;
		}
		if (testCfFile() == false)
		{
			System.err.println("Can not find CFs file!");
			return;
		}
		if (getCFNames() == false)
		{
			System.err.println("Can not read CFs correctly!");
			return;
		}
		if (ignoreSomeParaId && getIgnoreParaIds() == false)
		{
			System.err.println("Can not read ignoreParaIds correctly!");
			return;
		}
		
		try
		{
			Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
		}
		catch (ClassNotFoundException e)
		{
			e.printStackTrace();
			return;
		}
		catch (Exception e)
		{
			e.printStackTrace();
			return;
		}
		
		// set column family
		switch (schema)
		{
			case 1:
				// 工况表
				cfNameFormat = Config.CFNameFormat1;
				cfPrimaryKeyAlias = Config.CFPrimaryKeyAlias1;
				cQLInsertFormat = Config.CQLInsertFormat1;
				break;
			case 2:
				// 起重机表
				cfNameFormat = Config.CFNameFormat2;
				cfPrimaryKeyAlias = Config.CFPrimaryKeyAlias2;
				cQLInsertFormat = Config.CQLInsertFormat2;
				break;
			default:
				// 其他表
				cfNameFormat = Config.CFNameFormat0;
				cfPrimaryKeyAlias = Config.CFPrimaryKeyAlias0;
				cQLInsertFormat = Config.CQLInsertFormat0;
				break;
		}
		
		if (importType == ImportType.CLEAR_DATA)
		{
			if (dropTableWithThrift(keyspace) == false)
			{
				logger.info("Drop keyspace failed!");
				return;
			}
			return;
		}
		if (importType == ImportType.NEW_KEYSPACE)
		{
			if (createTableWithThrift(keyspace) == false)
			{
				logger.info("Create keyspace failed!");
				return;
			}
		}
		if (importType == ImportType.RECREATE_KEYSPACE)
		{
			if (dropTableWithThrift(keyspace) == false)
			{
				logger.info("Drop keyspace failed!");
			}
			if (createTableWithThrift(keyspace) == false)
			{
				logger.info("Create keyspace failed!");
				return;
			}
		}
		
		jedis = new Jedis(redisIp, 6379);
		jedis.select(16);
		jedis.set("threadnumber", String.valueOf(LaUDImport.thread));
		
		logger.info("Import program Daemon started!\n");
	}
	
	private static void doImport()
	{
		try
		{
			ProcessManager.InitProcessManager(logger, csvFolder, logFileFolder);
		}
		catch (Exception e)
		{
			System.err.println("Init ProcessManager failed: " + e.getMessage());
			return;
		}
		
		long starttime = System.nanoTime();
		
		logger.info("Create threads to import data to LaUD...");
		
		BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
		executor = new ThreadPoolExecutor(thread, thread, 100, TimeUnit.DAYS, queue);
		runners = new ArrayList<LaUDRunner>();
		mutationMapQueue = new ConcurrentLinkedQueue<MutationMap>();
		for (int i = 0; i < thread; i++)
		{
			LaUDRunner runner = new LaUDRunner(i, keyspace, mutationMapQueue);
			runners.add(runner);
			executor.execute(runner);
		}
		
		// main thread read csv
		Boolean waitAtFirst = true;
		while (halted == false)
		{
			// get source
			if (waitAtFirst)
			{
				logger.info("Wait to get available csv/gz source file...");
				waitAtFirst = false;
			}
			File file = ProcessManager.getAvailableSource();
			if (file == null)
			{
				if (nomoreCsv && LaUDImport.doNotWait)
				{
					logger.info("No more available csv/gz source file, thread will end.");
					break;
				}
				nomoreCsv = true;
				WaitSecond(10);
				continue;
			}
			waitAtFirst = true;
			String fileName = file.getAbsolutePath();
			logger.info("Find a new input file: " + fileName);
			// set start import time
			if (ProcessManager.importStartTime == 0)
			{
				ProcessManager.importStartTime = System.nanoTime();
			}
			// import
			boolean succeed = importFileData(file);
			// commit task
			if (succeed)
			{
				ProcessManager.commitTask(fileName, true);
				logger.info(String.format("Import file %s succeed!", fileName));
			}
			else
			{
				ProcessManager.commitTask(fileName, false);
				logger.info(String.format("Import file %s failed!", fileName));
				if (LaUDImport.errorDataStop)
				{
					halted = true;
					break;
				}
			}
		}
		ProcessManager.finishImport();
		
		// wait import threads to shutdown
		executor.shutdown();
		try
		{
			executor.awaitTermination(100, TimeUnit.DAYS);
		}
		catch (InterruptedException e1)
		{
			logger.error("Main thread interrupted while waiting " + e1.getMessage());
		}
		
		long endTime = System.nanoTime();
		logger.info("Import progress total time:"
				+ TimeUtils.nanoSecondToString((endTime - starttime)));
		
		logger.info("Import finish!");
	}
	
	private static boolean importFileData(File file)
	{
		timeSerialMap = null;
		System.gc();
		timeSerialMap = new HashMap<String, TimeSerial>(TIMESERIAL_MAP_INIT_CAPACITY);
		MutationMap mutationMap = new MutationMap();
		nowLineNum = 0;
		// nowImportCount = 0;
		BufferedReader bReader = null;
		boolean stopFlag = false;
		String fileName = file.getAbsolutePath();
		nowFileName = fileName;
		try
		{
			try
			{
				bReader = new BufferedReader(new FileReader(file));
				bReader.readLine();
			}
			catch (FileNotFoundException e3)
			{
				logger.error("Get BufferedReader failed: " + e3.getMessage());
				return false;
			}
			long time = System.nanoTime();
			try
			{
				List<String> lineDataList = new ArrayList<String>();
				int listDataNum = 0;
				boolean dataFlag = true;
				while (dataFlag && !stopFlag)
				{
					if (LaUDImport.halted)
					{
						bReader.close();
						return false;
					}
					String lineData = bReader.readLine();
					if (lineData != null)
					{
						listDataNum++;
						lineDataList.add(lineData);
						if (listDataNum < READ_BATCH_SIZE)
						{
							continue;
						}
					}
					else
					{
						dataFlag = false;
					}
					listDataNum = 0;
					List<RowData> rowDataList = new ArrayList<RowData>();
					for (String line : lineDataList)
					{
						RowData rowData = null;
						try
						{
							rowData = getRowData(line);
						}
						catch (DataFormatErrorException e)
						{
							if (LaUDImport.errorDataStop)
							{
								stopFlag = true;
								break;
							}
						}
						if (rowData != null)
						{
							rowDataList.add(rowData);
						}
						nowLineNum++;
					}
					lineDataList.clear();
					for (RowData rowData : rowDataList)
					{
						addRowData(rowData, mutationMap);
						
						if (mutationMap.mutationCount >= MUTATION_BATCH_SIZE)
						{
							while (mutationMapQueue.size() > thread * 10)
							{
								WaitSecond(1);
							}
							mutationMapQueue.add(mutationMap);
							mutationMap = new MutationMap();
						}
						
						ProcessManager.updateFileCount(fileName, 1, nowLineNum, 0);
					}
				}
			}
			catch (IOException e1)
			{
				stopFlag = true;
				logger.error("Error when imort file " + fileName + ", message : " + e1.getMessage());
			}
			catch (UnConfiguredCFException e2)
			{
				stopFlag = true;
				LaUDImport.halted = true;
			}
			finally
			{
				try
				{
					bReader.close();
					// finish success
					if (stopFlag == false)
					{
						if (mutationMap.mutationCount > 0)
						{
							mutationMapQueue.add(mutationMap);
						}
						ProcessManager.updateFileCount(fileName, 2, nowLineNum, nowLineNum);
					}
				}
				catch (IOException e)
				{
					stopFlag = true;
					e.printStackTrace();
				}
				catch (NullPointerException e1)
				{
					stopFlag = true;
					e1.printStackTrace();
					logger.error(e1.getMessage());
				}
			}
			logger.info(String.format("File import %s successful. Size:%s; Time used:%s",
					file.getName(), String.valueOf(nowLineNum),
					TimeUtils.nanoSecondToString(System.nanoTime() - time)));
		}
		catch (Throwable t)
		{
			stopFlag = true;
			t.printStackTrace();
			logger.error(t.getMessage());
		}
		
		return (stopFlag == false);
	}
	
	private static void addRowData(RowData rowData, MutationMap mutationMap)
			throws UnConfiguredCFException
	{
		Column column = new Column(ByteBufferUtil.bytes(rowData.localTime));
		column.setValue(ByteBufferUtil.bytes(rowData.paravalue));
		column.setTimestamp(System.currentTimeMillis());
		Mutation mutation = new Mutation();
		ColumnOrSuperColumn colOrSuperCol = new ColumnOrSuperColumn();
		colOrSuperCol.column = column;
		mutation.column_or_supercolumn = colOrSuperCol;
		ByteBuffer keyBuffer = ByteBufferUtil.bytes(rowData.equipCode);
		String cfName = String.format(LaUDImport.cfNameFormat, rowData.paraId);
		mutationMap.addMutation(keyBuffer, cfName, mutation);
	}
	
	private static RowData getRowData(String lineString) throws DataFormatErrorException
	{
		String equipCode = null;
		int paraId = 0;
		String paravalue = null;
		Date localDateTime = null;
		try
		{
			String[] data = lineString.split(",");
			if (data.length != 6)
			{
				ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
				return null;
			}
			equipCode = trimData(data[1]);
			paraId = Integer.valueOf(trimData(data[2]));
			if (LaUDImport.ignoreSomeParaId && LaUDImport.ignoreParaIds.contains(paraId))
			{
				return null;
			}
			paravalue = trimData(data[3]);
			String dataString = trimData(data[5]);
			try
			{
				localDateTime = sdf.parse(dataString);
			}
			catch (ParseException e)
			{
				try
				{
					localDateTime = sdf2.parse(dataString);
				}
				catch (ParseException e1)
				{
					ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
					throw new DataFormatErrorException();
				}
			}
		}
		catch (NumberFormatException e1)
		{
			ProcessManager.errorDataLog(nowFileName, nowLineNum, lineString);
			throw new DataFormatErrorException();
		}
		// 添加流水号
		long localTime = localDateTime.getTime();
		// String serialKey = getSerialKey(equipCode, paraId, localDateTime);
		String serialKey = equipCode + paraId;
		TimeSerial timeSerial = timeSerialMap.get(serialKey);
		if (timeSerial == null)
		{
			timeSerial = new TimeSerial(localTime, 0);
			timeSerialMap.put(serialKey, timeSerial);
		}
		else
		{
			if (localTime != timeSerial.time)
			{
				timeSerial.time = localTime;
				timeSerial.serial = 0;
			}
			else
			{
				timeSerial.serial++;
				localTime += timeSerial.serial;
				if (timeSerial.serial >= 1000)
				{
					ProcessManager.dataTooManyLog(nowFileName, nowLineNum, lineString);
				}
			}
		}
		RowData rowData = new RowData();
		rowData.paraId = paraId;
		rowData.paravalue = paravalue;
		rowData.equipCode = equipCode;
		rowData.localTime = localTime;
		return rowData;
	}
	
	private static String trimData(String oriData)
	{
		oriData = oriData.trim();
		oriData = oriData.substring(1, oriData.length() - 1);
		return oriData;
	}
	
	@SuppressWarnings("unused")
	private static String getSerialKey(String equipCode, int paraId, Date localDateTime)
	{
		Calendar calendar = Calendar.getInstance();
		calendar.setTime(localDateTime);
		StringBuilder stringBuilder = new StringBuilder(20);
		stringBuilder.append(equipCode);
		stringBuilder.append(String.valueOf(paraId));
		int hour = calendar.get(Calendar.HOUR_OF_DAY);
		int minute = calendar.get(Calendar.MINUTE);
		int second = calendar.get(Calendar.SECOND);
		int millisecond = calendar.get(Calendar.MILLISECOND);
		int allMillisecond = (((hour * 60) + minute) * 60 + second) * 1000 + millisecond;
		stringBuilder.append(String.valueOf(allMillisecond));
		return stringBuilder.toString();
	}
	
	private static void WaitSecond(int secound)
	{
		try
		{
			Thread.sleep(secound * 1000);
		}
		catch (InterruptedException e)
		{
		}
	}
	
	protected static void parseParams(String[] args)
	{
		for (int i = 0; i < args.length; i += 2)
		{
			if (args[i].equals("--thread"))
			{
				thread = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--schema"))
			{
				schema = Integer.valueOf(args[i + 1]);
				if (keyspace == null)
				{
					keyspace = "laud_sany_schema" + schema;
				}
			}
			else if (args[i].equals("--type"))
			{
				String valString = args[i + 1];
				if (valString.equals("new"))
				{
					importType = ImportType.NEW_KEYSPACE;
				}
				else if (valString.equals("clear"))
				{
					importType = ImportType.CLEAR_DATA;
				}
				else if (valString.equals("recreate"))
				{
					importType = ImportType.RECREATE_KEYSPACE;
				}
				else
				{
					importType = ImportType.APPEND_DATA;
				}
			}
			else if (args[i].equals("--laudport"))
			{
				laudPort = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--laudip"))
			{
				IP = args[i + 1].split(",");
				host = new String[IP.length];
				for (int k = 0; k < IP.length; k++)
				{
					host[k] = String.format(Config.JDBCFormat, IP[k], laudPort);
				}
			}
			else if (args[i].equals("--replica"))
			{
				replicationFactor = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--keyspace"))
			{
				keyspace = args[i + 1];
			}
			else if (args[i].equals("--csvpath"))
			{
				csvFolder = args[i + 1];
			}
			else if (args[i].equals("--cfsfile"))
			{
				cfsFile = args[i + 1];
			}
			else if (args[i].equals("--ignorefile"))
			{
				ignoreSomeParaId = true;
				ignoreParaIdsFile = args[i + 1];
			}
			else if (args[i].equals("--errorstop"))
			{
				if (args[i + 1].equals("1"))
				{
					errorDataStop = true;
				}
			}
			else if (args[i].equals("--comparator"))
			{
				cfComparatorType = Config.CFComparatorTypePrefix + args[i + 1];
			}
			else if (args[i].equals("--transcation"))
			{
				writeTranscation = args[i + 1];
				if (writeTranscation.equals("ANY"))
				{
					LaUDRunner.CL = ConsistencyLevel.ANY;
				}
				else if (writeTranscation.equals("ALL"))
				{
					LaUDRunner.CL = ConsistencyLevel.ALL;
				}
				else if (writeTranscation.equals("QUORUM"))
				{
					LaUDRunner.CL = ConsistencyLevel.QUORUM;
				}
			}
			else if (args[i].equals("--donotwait"))
			{
				if (args[i + 1].equals("1"))
				{
					doNotWait = true;
				}
			}
			else if (args[i].equals("--redisip"))
			{
				redisIp = args[i + 1];
			}
		}
		logger.info("Parameters are:");
		logger.info("thread = " + thread);
		logger.info("schema = " + schema);
		logger.info("type = " + importType);
		logger.info("laud port = " + laudPort);
		String ipsString = "";
		for (String ip : IP)
		{
			ipsString += ip + "; ";
		}
		logger.info("laud ip = " + ipsString);
		logger.info("replication factor = " + replicationFactor);
		logger.info("keyspace = " + keyspace);
		logger.info("csv path = " + csvFolder);
		logger.info("cfs file = " + cfsFile);
		logger.info("ignore file = " + ignoreParaIdsFile);
		logger.info("errorData Stop = " + errorDataStop);
		logger.info("comparator = " + cfComparatorType);
		logger.info("transcation = " + LaUDRunner.CL);
		logger.info("doNotWait = " + doNotWait);
		logger.info("redisIp = " + redisIp);
	}
	
	protected static boolean getCFNames()
	{
		boolean flag = true;
		File file = new File(cfsFile);
		BufferedReader bReader = null;
		String paraIdString = null;
		ArrayList<Integer> cfList = new ArrayList<Integer>();
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((paraIdString = bReader.readLine()) != null)
			{
				paraIdString = paraIdString.trim();
				if (paraIdString.equals(""))
				{
					continue;
				}
				int paraId = Integer.valueOf(paraIdString);
				cfList.add(paraId);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("Can not open cfs file");
		}
		catch (NumberFormatException e2)
		{
			flag = false;
			logger.error("Cfs file contains invalid integer");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("Get cfs failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		if (flag == true)
		{
			cFS = cfList.toArray(new Integer[0]);
		}
		return flag;
	}
	
	protected static boolean testCsvFoler()
	{
		try
		{
			File file = new File(csvFolder);
			if (file.exists())
			{
				return true;
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		return false;
	}
	
	protected static boolean testCfFile()
	{
		try
		{
			File file = new File(cfsFile);
			if (file.exists())
			{
				return true;
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		return false;
	}
	
	protected static boolean getIgnoreParaIds()
	{
		ignoreParaIds = new HashSet<Integer>();
		boolean flag = true;
		File file = new File(ignoreParaIdsFile);
		BufferedReader bReader = null;
		String paraIdString = null;
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((paraIdString = bReader.readLine()) != null)
			{
				paraIdString = paraIdString.trim();
				if (paraIdString.equals(""))
				{
					continue;
				}
				int paraId = Integer.valueOf(paraIdString);
				ignoreParaIds.add(paraId);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("Can not open ignoreParaIds file!");
		}
		catch (NumberFormatException e2)
		{
			flag = false;
			logger.error("IgnoreParaIds file contains invalid integer!");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("Get ignoreParaIds failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		return flag;
	}
	
	protected static boolean dropTableWithThrift(String keyspace) throws IOException
	{
		boolean flag = true;
		logger.info("Waiting server to drop keyspace...");
		TTransport tr = new TFramedTransport(new TSocket(IP[0], laudPort));
		TProtocol proto = new TBinaryProtocol(tr);
		Cassandra.Client client = new Cassandra.Client(proto);
		try
		{
			tr.open();
			client.system_drop_keyspace(keyspace);
			logger.info("Drop keyspace successful!");
			logger.info("Wait for 10 sec to take effert.");
			try
			{
				Thread.sleep(10000);
			}
			catch (InterruptedException e)
			{
				e.printStackTrace();
			}
		}
		catch (Exception e)
		{
			flag = false;
			logger.error("Can not drop keyspace, IP:" + IP[0] + " Port:" + laudPort);
			logger.error("Caused by: " + e.toString());
		}
		finally
		{
			if (tr != null)
			{
				tr.close();
			}
		}
		return flag;
	}
	
	protected static boolean createTableWithThrift(String keyspace)
	{
		logger.info("Waiting server to create keyspace and schema...");
		boolean flag = true;
		TTransport tr = new TFramedTransport(new TSocket(IP[0], laudPort));
		TProtocol proto = new TBinaryProtocol(tr);
		Cassandra.Client client = new Cassandra.Client(proto);
		List<CfDef> cfDefs = new ArrayList<CfDef>();
		try
		{
			tr.open();
			for (int paraId : cFS)
			{
				CfDef cfDef = new CfDef(keyspace, String.format(cfNameFormat,
						String.valueOf(paraId)));
				cfDef.setKey_alias(cfPrimaryKeyAlias.getBytes());
				cfDef.setComparator_type(cfComparatorType);
				cfDef.setDefault_validation_class(Config.CFCalidationClass);
				cfDef.setKey_validation_class(Config.CFCalidationClass);
				cfDefs.add(cfDef);
			}
			KsDef ks_def = new KsDef(keyspace, Config.KSStrategy, cfDefs);
			ks_def.setStrategy_options(Collections.singletonMap("replication_factor",
					String.valueOf(replicationFactor)));
			client.system_add_keyspace(ks_def);
			logger.info("Create keyspace and schema successful!");
			logger.info("Wait for 30 sec to take effert.");
			try
			{
				Thread.sleep(30000);
			}
			catch (InterruptedException e)
			{
				e.printStackTrace();
			}
			
		}
		catch (Exception e)
		{
			flag = false;
			logger.error("Can not connect create keyspace schema, IP:" + IP[0] + " Port:"
					+ laudPort);
			logger.error("Caused by: " + e.toString());
		}
		finally
		{
			if (tr != null)
			{
				tr.close();
			}
		}
		return flag;
	}
	
	public static void stopProgram()
	{
		halted = true;
		logger.info("Try to stop import program...");
		for (LaUDRunner runner : runners)
		{
			executor.remove(runner);
			runner.interrupt();
		}
		System.exit(1);
	}
	
}
